package org.hong.monkey.serializer

import java.io.{InputStream, EOFException, OutputStream}
import java.nio.ByteBuffer
import javax.annotation.concurrent.NotThreadSafe

import org.hong.monkey.MonkeyEnv
import org.hong.monkey.annotation.{Private, DeveloperApi}
import org.hong.monkey.util.NextIterator

import scala.reflect.ClassTag

@DeveloperApi
abstract class Serializer {
  @volatile protected var defaultClassLoader: Option[ClassLoader] = None

  def setDefaultClassLoader(classLoader: ClassLoader): Serializer = {
    defaultClassLoader = Some(classLoader)
    this
  }

  def newInstance(): SerializerInstance

  @Private
  private[monkey] def supportsRelocationOfSerializedObjects:Boolean = false
}

@DeveloperApi
object Serializer {
  def getSerializer(serializer: Serializer): Serializer = {
    if (serializer == null) MonkeyEnv.get.serializer else serializer
  }

  def getSerializer(serializer: Option[Serializer]): Serializer = {
    serializer.getOrElse(MonkeyEnv.get.serializer)
  }
}

@DeveloperApi
@NotThreadSafe
abstract class SerializerInstance {
  def serialize[T: ClassTag](t: T): ByteBuffer

  def deserialize[T: ClassTag](bytes: ByteBuffer): T

  def deserialize[T: ClassTag](bytes: ByteBuffer, loader: ClassLoader): T

  def serializeStream(s: OutputStream): SerializationStream

  def deserializeStream(s: InputStream): DeserializationStream
}

@DeveloperApi
abstract class SerializationStream {
  def writeObject[T: ClassTag](t: T): SerializationStream
  def writeKey[T: ClassTag](key: T): SerializationStream = writeObject(key)
  def writeValue[T: ClassTag](value: T): SerializationStream = writeObject(value)
  def flush(): Unit
  def close(): Unit

  def writeAll[T: ClassTag](iter: Iterator[T]): SerializationStream = {
    while (iter.hasNext) {
      writeObject(iter.next())
    }
    this
  }
}

@DeveloperApi
abstract class DeserializationStream {
  def readObject[T: ClassTag](): T
  def readKey[T: ClassTag](): T = readObject[T]()
  def readValue[T: ClassTag](): T = readObject[T]()
  def close(): Unit

  def asIterator: Iterator[Any] = new NextIterator[Any] {
    override protected def getNext() = {
      try {
        readObject[Any]()
      } catch {
        case eof: EOFException =>
          finished = true
          null
      }
    }

    override protected def close(): Unit = {
      DeserializationStream.this.close()
    }
  }

  def asKeyValueIterator: Iterator[(Any, Any)] = new NextIterator[(Any, Any)] {
    override protected def getNext() = {
      try {
        (readKey[Any](), readValue[Any]())
      } catch {
        case eof: EOFException =>
          finished = true
          null
      }
    }

    override protected def close(): Unit = {
      DeserializationStream.this.close()
    }
  }
}
